package com.xrui.hbase.codec.impl;

import com.google.common.base.Preconditions;
import com.google.protobuf.AbstractMessageLite;
import com.xrui.hbase.DecodedCell;
import com.xrui.hbase.codec.CellDecoder;
import com.xrui.hbase.impl.BoundColumnReaderSpec;
import com.xrui.hbase.schema.SchemaType;
import com.xrui.hbase.schema.TableLayout;
import com.xrui.hbase.schema.impl.ProtobufCellSchema;
import com.xrui.hbase.schema.spec.CellSpec;
import com.xrui.hbase.schema.spec.ColumnReaderSpec;


import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;

/**
 * De-serializes an HBase cell encoded as a Protocol Buffer.
 *
 * @param <T> The type of the decoded cell data. Should be a Protocol Buffer generated class.
 */
public final class ProtobufCellDecoder<T> implements CellDecoder<T> {
    /* The name of this provider */
    public static final String PROVIDER_NAME = "ProtobufCellDecoder";

    /**
     * Class of the protocol buffer to decode.
     */
    private final Class<? extends AbstractMessageLite> mProtoClass;

    /**
     * 'parseFrom(byte[] bytes)' static method of the protocol buffer class to decode.
     */
    private final Method mParseFromMethod;

    // -----------------------------------------------------------------------------------------------

    /**
     * Initializes a ProtobufCellDecoder.
     *
     * @param cellSpec Specification of the cell encoding.
     * @throws IOException on I/O error.
     */
    public ProtobufCellDecoder(CellSpec cellSpec) throws IOException {
        Preconditions.checkNotNull(cellSpec);
        Preconditions.checkArgument(
            cellSpec.getCellSchema().getType() == SchemaType.PROTOBUF);
        ProtobufCellSchema cellSchema = (ProtobufCellSchema) cellSpec.getCellSchema();
        final String className = cellSchema.getProtobufClassName();
        try {
            mProtoClass = Class.forName(className).asSubclass(AbstractMessageLite.class);
        } catch (ClassNotFoundException cnfe) {
            throw new IOException(cnfe);
        }

        try {
            mParseFromMethod = mProtoClass.getMethod("parseFrom", byte[].class);
        } catch (NoSuchMethodException nsme) {
            throw new IOException(nsme);
        }
    }

    /**
     * Initializes a ProtobufCellDecoder.
     *
     * @param layout TableLayout from which to create the Protobuf class name.
     * @param spec   Specification of the cell encoding.
     * @throws IOException on I/O error.
     */
    public ProtobufCellDecoder(TableLayout layout, BoundColumnReaderSpec spec)
        throws IOException {
        Preconditions.checkNotNull(layout);
        Preconditions.checkNotNull(spec);
        Preconditions.checkArgument(
            spec.getColumnReaderSpec().getEncoding() == ColumnReaderSpec.Encoding.PROTOBUF);
        ProtobufCellSchema cellSchema = (ProtobufCellSchema) layout.getCellSchema(spec.getColumnName());
        final String className = cellSchema.getProtobufClassName();
        try {
            mProtoClass = Class.forName(className).asSubclass(AbstractMessageLite.class);
        } catch (ClassNotFoundException cnfe) {
            throw new IOException(cnfe);
        }

        try {
            mParseFromMethod = mProtoClass.getMethod("parseFrom", byte[].class);
        } catch (NoSuchMethodException nsme) {
            throw new IOException(nsme);
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    @SuppressWarnings("unchecked")
    public DecodedCell<T> decodeCell(byte[] encodedBytes) throws IOException {
        try {
            return new DecodedCell<>((T) mParseFromMethod.invoke(mProtoClass, (Object) encodedBytes));
        } catch (InvocationTargetException | IllegalAccessException ite) {
            throw new IOException(ite);
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public T decodeValue(byte[] bytes) throws IOException {
        return decodeCell(bytes).getData();
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public String getName() {
        return PROVIDER_NAME;
    }
}
